package net.mingsoft.report_data.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.reflect.ReflectDatumReader;
import org.apache.avro.specific.SpecificRecordBase;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.List;

/**
 * Kafka序列穿传输avro格式处理类
 * @author Administrator
 */
@Slf4j
public class AvroUtil {

    /**
     * 序列化发送数据为byte
     */
    public static <T extends SpecificRecordBase> byte[] serialize(T postData) {
        byte[] bytes;
        try {
            // 初始化序列化工具 size: 10MB
            ByteArrayOutputStream out = new ByteArrayOutputStream(10485760);
            BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
            GenericDatumWriter<GenericRecord> writer = new GenericDatumWriter<>(postData.getSchema());
            // 转成kafka需要的格式
            GenericRecord payload = new GenericData.Record(postData.getSchema());
            List<Schema.Field> fields = postData.getSchema().getFields();
            for (int i = 0; i < fields.size(); i++) {
                payload.put(i, postData.get(fields.get(i).name()));
            }
            writer.write(payload, encoder);
            encoder.flush();
            out.flush();
            bytes = out.toByteArray();
        } catch (IOException e) {
            log.error("序列化avro数据失败：", e);
            throw new RuntimeException("序列化avro数据失败");
        }

        log.info("发送数据length：{}", bytes.length);
        return bytes;
    }

    /**
     * 反序列化对象
     */
    public static <T extends SpecificRecordBase> T deserialize(byte[] bytes, Class<T> tClass) {
        T result;
        try {
            result = tClass.newInstance();
        } catch (Exception e) {
            log.error("构造泛型类失败", e);
            throw new RuntimeException("构造泛型类失败");
        }

        // 反序列化数据
        Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, null);
        ReflectDatumReader<T> reflectDatumReader = new ReflectDatumReader<>(tClass);
        try {
            reflectDatumReader.read(result, decoder);
            log.info("反序列化avro数据:{}", result);
            return result;
        } catch (IOException e) {
            log.error("反序列化数据失败:", e);
            throw new RuntimeException("反序列化数据失败");
        }
    }
}
